package com.sunday.common.mq.rabbit.study.base.e10_QueueLengthLimit;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.sunday.common.mq.rabbit.study.base.ConnectionUtil;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;

import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.HashMap;
import java.util.concurrent.TimeUnit;

/**
 * 队列长度限制
 * 队列的最大长度可以限制为设定的 消息，或一组字节数（所有消息的总和） 正文长度，忽略消息属性和任何开销），或 双。
 * 对于任何给定的队列，最大长度（任一类型）可以是 使用策略定义（强烈建议使用此选项） 或由客户端使用队列的可选参数。 在有效队列策略和参数都指定最大长度的情况下， 将使用两个值中的最小值。
 * 队列长度设置也可以由操作员策略强制执行。
 * 在所有情况下，都使用处于就绪状态的消息数;消费者未确认的消息不计入限制。
 * 可以观察到就绪消息的数量及其占用空间（以字节为单位） 使用 Rabbitmqctl list_queues输出中的messages_ready和message_bytes_ready，以及管理 UI 和 HTTP API 响应中类似名称的字段。
 */
@Slf4j
public class QueueLengthLimit {

    /**
     * 默认最大队列长度限制行为
     * RabbitMQ 在最大队列长度或 设置大小并达到最大值是从前面丢弃或死信消息 队列（即队列中最早的消息）。要修改 此行为，请使用下面描述的溢出设置。
     * <p>
     * 队列溢出行为
     * 使用溢出设置配置队列溢出 行为。如果溢出设置为 reject-publish 或 reject-publish-dlx， 最近发布的消息将被丢弃。
     * 此外，如果发布者确认已启用，则发布者将通过 basic.nack 消息收到拒绝通知。
     * 如果邮件被路由到多个 队列并被其中至少一个拒绝，通道将通知 出版商通过basic.nack。
     * 消息仍将是 发布到可以排队的所有其他队列。 reject-publish 和 reject-publish-dlx 之间的区别在于，reject-publish-dlx 也是死信拒绝消息。
     * <p>
     * 在声明期间使用 x 参数定义最大队列长度
     * 可以通过向 x-max-length 队列声明参数提供带有 非负整数值。
     * 可以通过向 x max-length-bytes 队列声明参数提供带有 非负整数值。（当队列的内存达到指定字节时，将采用LRU算法对以往消息进行删除）
     * 如果两个参数都设置了，那么两者都将适用;无论哪个限制 先命中将强制执行。
     * 可以通过向 x-overflow 队列声明参数提供 字符串值。可能的值为 drop-head （默认值）、reject-publish 或 reject-publish-dlx。
     * <p>
     * x-overflow -> queue溢出行为，这将决定当队列达到设置的最大长度或者最大的存储空间时发送到消息队列的消息的处理方式；
     * 有效的值是：
     * drop-head（删除queue头部的消息）、
     * reject-publish（最近发来的消息将被丢弃）、
     * reject-publish-dlx（拒绝发送消息到死信交换器）
     * 类型为quorum 的queue只支持drop-head;
     */
    @Test
    @SneakyThrows
    public void defaultMaxQueueLengthLimitBehaviour() {

        ConnectionFactory cf = ConnectionUtil.factory();
        try (Connection connection = cf.newConnection()) {
            Channel channel = connection.createChannel();

            channel.queueDelete("queue_length");
            channel.queueDelete("queue_length_bytes");

            /**
             * x-max-length 设置队列中可以存储处于ready状态消息的数量
             */
            channel.queueDeclare("queue_length", false, false, false, Collections.singletonMap("x-max-length", 5));

            for (int i = 0; i < 8; i++) {
                channel.basicPublish("", "queue_length", null, String.valueOf(i).getBytes(StandardCharsets.UTF_8));
            }

            channel.basicConsume("queue_length", true,
                    (consumerTag, delivery) -> {
                        String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                        log.info(" [x] Received '{}'", message);
                    },
                    consumerTag -> {
                    });

            TimeUnit.SECONDS.sleep(1);

            log.info("==============================================");


            /**
             * x-max-length-bytes 队列中可以存储处于ready状态消息占用内存的大小(只计算消息体的字节数，不计算消息头、消息属性占用的字节数)
             */
            channel.queueDeclare("queue_length_bytes", false, false, false,
                    new HashMap<String, Object>() {{
                        put("x-max-length-bytes", 2);
//                        put("x-overflow", "drop-head"); // drop-head（删除queue头部的消息）缺省
                        put("x-overflow", "reject-publish"); // reject-publish（最近发来的消息将被丢弃）
//                        put("x-overflow", "reject-publish-dlx"); // reject-publish-dlx（拒绝发送消息到死信交换器）
                    }});

            for (int i = 0; i < 8; i++) {
                channel.basicPublish("", "queue_length_bytes", null, String.valueOf(i).getBytes(StandardCharsets.UTF_8));
            }

            channel.basicConsume("queue_length_bytes", true,
                    (consumerTag, delivery) -> {
                        String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                        log.info(" [bytes] Received '{}'", message);
                    },
                    consumerTag -> {
                    });

            TimeUnit.SECONDS.sleep(1);

            log.info("==============================================");

            /**
             * 本轮测试可知 :
             * 当满足 x-max-length 或者 x-max-length-bytes
             * 并且根据 x-overflow 配置，清理信息
             */

        }
    }

}
